package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.*;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.concurrent.Future;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Created by tang on 2018/12/16.
 */
public class StatusManagement extends BusiProcessing {
    Logger logger = Logger.getLogger(StatusManagement.class);

    private static final String GROUP_SERVICE_URL = "bungarus.group-service.url";
    private static final String GROUP_SERVICE_TIMEOUT = "bungarus.group-service.timeout";
    private static final String GROUP_SERVICE_RETRY_TIMES = "bungarus.group-service.retry.times";
    private static final String GROUP_SERVICE_REDIS_TEMPLATE = "groupServiceRedisTemplate";

    private static final String BASE_RES = "tenants/";
    private static final int GROUP_PAGE_SIZE = 50;
    private static final String PREFIX_GROUP = "group:";

    private static StatusManagement statusManagement = new StatusManagement();

    private CloseableHttpClient client;

    private RedisTemplate<String, Object> groupMemberCache;

    private String url;
    private int timeout;
    private int retryTotalCount = 0;

    private Busi busiServer;

    private StatusManagement() {
        init();
    }

    private void init() {
        this.url = SessionManagement.springCtx.getEnvironment().getProperty(GROUP_SERVICE_URL);
        assert this.url.length() > 0;
        if(!this.url.endsWith("/")) {
            this.url = this.url + "/";
        }
        this.url = this.url + BASE_RES;
        this.timeout = Integer.valueOf(SessionManagement.springCtx.getEnvironment().getProperty(GROUP_SERVICE_TIMEOUT));

        String retry = SessionManagement.springCtx.getEnvironment().getProperty(GROUP_SERVICE_RETRY_TIMES);
        if(!StringUtils.isEmpty(retry)) {
            this.retryTotalCount = Integer.valueOf(retry);
        }

        PoolingHttpClientConnectionManager clientConnectionManager = new PoolingHttpClientConnectionManager();
        clientConnectionManager.setMaxTotal(Integer.MAX_VALUE);
        clientConnectionManager.setDefaultMaxPerRoute(Integer.MAX_VALUE);

        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectionRequestTimeout(timeout)
                .setConnectTimeout(timeout)
                .setSocketTimeout(timeout)
                .build();

        client = HttpClients.custom()
                    .setKeepAliveStrategy((HttpResponse response, HttpContext context) -> {
                        HeaderElementIterator it = new BasicHeaderElementIterator
                                (response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                        while (it.hasNext()) {
                            HeaderElement he = it.nextElement();
                            String param = he.getName();
                            String value = he.getValue();
                            if (value != null && param.equalsIgnoreCase
                                    ("timeout")) {
                                return Long.parseLong(value) * 1000;
                            }
                        }
                        return 5 * 1000;
                    })
                    .setConnectionManager(clientConnectionManager)
                    .setDefaultRequestConfig(requestConfig)
                    .build();

        groupMemberCache = (RedisTemplate) SessionManagement.springCtx.getBean(GROUP_SERVICE_REDIS_TEMPLATE);
    }

    public static StatusManagement getInstance() {
        return statusManagement;
    }

    @Override
    public void doBusiness(Object o) throws BusiProcessingException {
        if(null == o || !(o instanceof Map)) {
            throw new IllegalArgumentException("Input argument must be instance of Map.");
        }
        Map map = (Map)o;
        busiServer = (Busi)map.get(SessionManagement.BUSI_SERVER);

        assert busiServer != null;

        ChannelHandlerContext ctx = (ChannelHandlerContext)map.get(SESSION_CONTEXT);
        MessageProtobuf.Message statusMsg = (MessageProtobuf.Message)map.get(SESSION);

        String to = statusMsg.getHeader().getTo();
        int status;
        if(MessageType.AUTH == statusMsg.getHeader().getType()) {
            status = OnlineStatus.ONLINE;
        }
        else { //The message type is MessageType.STATUS
            status = OnlineStatus.IDLE;
            for(MessageProtobuf.Extension extension: statusMsg.getHeader().getExtensionsList()) {
                if(extension.getKey().equals(MessageType.STATUS)) {
                    status = Integer.valueOf(extension.getValue());
                    break;
                }
            }
        }
        try {
            doStatus(ctx, statusMsg, status);
        }
        catch (StatusContainerException e) {
            e.printStackTrace();
        }
    }

    private void doStatus(ChannelHandlerContext ctx, MessageProtobuf.Message statusMsg, int status) throws StatusContainerException{
        //update the user's status in status container
        StatusContainer.getInstance().put(statusMsg.getHeader().getTenantID() + ":" + statusMsg.getHeader().getFrom(), status);
        //broadcast the status to everyone who is in the same group with him.
        int offset = 0, retried = 0;
        PagePayload<Group> groupPage = getGroupsMemberBelongstoWithRetry(
                retried,
                statusMsg.getHeader().getTenantID(),
                statusMsg.getHeader().getFrom(),
                statusMsg.getHeader().getToken(),
                offset);

        if(null != groupPage) {
            if ((groupPage.getPage() + 1) * groupPage.getSize() >= groupPage.getTotalCount()) {
                //Only one page. The size of groups the user belongs to is less than GROUP_PAGE_SIZE = 50.
                notifyMembersNewStatus(groupPage.getPageContents(), statusMsg, status);
            }
            else {
                do {
                    notifyMembersNewStatus(groupPage.getPageContents(), statusMsg, status);
                    groupPage.getPageContents().clear();
                    offset++;
                    groupPage = getGroupsMemberBelongstoWithRetry(
                            retried,
                            statusMsg.getHeader().getTenantID(),
                            statusMsg.getHeader().getFrom(),
                            statusMsg.getHeader().getToken(),
                            offset);

                }
                while (null != groupPage && groupPage.getPageContents().size() > 0);
            }
            //send status's ack message back to the sender
            ctx.writeAndFlush(MessageBuilder.buildAckMessage(statusMsg.getHeader().getId()));
        }
    }

    private void notifyMembersNewStatus(List<Group> pageContents, MessageProtobuf.Message msg, int status) {
        Map<String, InnerClient> accessClients = busiServer.getInnerClients();
        assert accessClients.size() > 0;
        //get member id list from groupMember cache
        int size = 200;
        BusiProcessing.busiTaskPool.execute(() -> {
            pageContents.forEach(group -> {
                int i = 0;
                Set<Object> members = null;
                try {
                    members = this.groupMemberCache.opsForZSet().range(
                            PREFIX_GROUP + group.getTenantId() + ":" + group.get_id(), i, i + size - 1);
                    while (members.size() > 0) {
                        //get member's session from session container
                        List<Object> sessions = SessionManagement.getInstanct().retrieveM(members, String.valueOf(group.getTenantId()));
                        //send status message to access node from which the user connected
                        sessions.stream().forEach(session -> {
                            Set userSession = (Set) session;
                            sendToAccess(userSession, msg, status);
                        });
                        sessions.clear();
                        sessions = null;
                        members.clear();
                        members = null;

                        i = i + size;
                        members = this.groupMemberCache.opsForZSet().range(
                                PREFIX_GROUP + group.getTenantId() + ":" + group.get_id(), i, i + size);
                    }
                }
                catch (Exception e){
                    e.printStackTrace();
                }
                finally {
                    if(null != members) {
                        members.clear();
                        members = null;
                    }
                }
            });
        });
    }

    private void sendToAccess(Set userSession, MessageProtobuf.Message msg, int status) {
        userSession.stream().forEach(s -> {
            //value pattern: {userId},{sessionId},{ipport}
            String[] values = StringUtils.tokenizeToStringArray(s.toString(), SessionManagement.VALUE_SEPERATOR);
            String reciptId = values[0];
            String sessionId = values[1];
            String ipport = values[2];

            List<InnerClient> accessClients = this.busiServer.getInnerClientList().get(Busi.CLUSTER_TAG_ACCESS);
            for(InnerClient accessClient: accessClients) {
                if(ipport.equals(accessClient.getSrvIp() + ":" + accessClient.getSrvPort())) {
                    try {
                        MessageProtobuf.Message mmsg = MessageBuilder.buildStatusMessage(sessionId, reciptId, msg, status);
                        accessClient.sendMessage(mmsg, Busi.CLUSTER_TAG_ACCESS);
                    }
                    catch (BungarusClientException e) {
                        e.printStackTrace();
                    }
                    break;
                }
            }
        });
    }

    private PagePayload<Group> getGroupsMemberBelongstoWithRetry(int retryCounts, int tenantId,
                                                                 String userId, String token, int offset)
            throws StatusContainerException {
        StringBuilder getEndpoint = new StringBuilder(this.url);
        getEndpoint.append(tenantId);
        getEndpoint.append("/");
        getEndpoint.append("groups?mid=");
        getEndpoint.append(userId);
        getEndpoint.append("&size=");
        getEndpoint.append(GROUP_PAGE_SIZE);
        getEndpoint.append("&offset=");

        Future<PagePayload<Group>> futureResult = BusiProcessing.busiTaskPool.submit(() -> {
            PagePayload<Group> page = null;
            HttpGet getMethod = new HttpGet(getEndpoint.append(offset).toString());
            getMethod.setHeader(HttpRequestXHeader.X_BUNGARUS_TOKEN, token);
            try {
                CloseableHttpResponse response = client.execute(getMethod);
                HttpEntity entity = response.getEntity();
                if (entity.getContentType().getValue().contains("application/json")) {
                    String strEntity = EntityUtils.toString(entity, "utf-8");

                    ObjectMapper objectMapper = new ObjectMapper();
                    JavaType javaType = objectMapper.getTypeFactory().constructParametricType(PagePayload.class, Group.class);
                    page = objectMapper.readValue(strEntity, javaType);
                    return page;
                }
            }
            catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        });
        try {
            return futureResult.get(this.timeout, TimeUnit.MILLISECONDS);
        }
        catch (TimeoutException e) {
            e.printStackTrace();
            //retry when timeout happened.
            if(retryCounts < this.retryTotalCount) {
                retryCounts++;
                return getGroupsMemberBelongstoWithRetry(retryCounts, tenantId, userId, token, offset);
            }
        }
        catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            throw new StatusContainerException("can not retrieve group list contain the user : " + userId);
        }
        finally {
            if (futureResult.isCancellable()) {
                futureResult.cancel(true);
            }
        }
        return null;
    }
}
